<?php
/**
 * Created by PhpStorm.
 * User: ZCY
 * Date: 2018/11/26
 * Time: 18:01
 */

namespace common\service;
use PhpAmqpLib\Channel\AMQPChannel;
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
use yii\base\Component;

class RabbitMq extends Component
{
    public $username;
    public $passwd;
    public $host;
    public $port;
    public $vhost = '/';

    private $client;
    private $channel;

    /**
     * 获取队列client
     * @param bool $reset 是否重置
     * @return \PhpAmqpLib\Connection\AMQPStreamConnection
     */
    protected function getClient($reset = false)
    {
        if (!$this->client || $reset) {
            $this->client = new AMQPStreamConnection($this->host, $this->port, $this->username, $this->passwd);
        }
        return $this->client;
    }

    /**
     * 获取队列channel
     * @return \PhpAmqpLib\Channel\AMQPChannel
     */
    protected function getChannel()
    {
        /**
         * @var AMQPChannel $this->channel
         */
        if (empty($this->channel) || null === $this->channel->getChannelId()) {
            $this->channel = $this->getClient()->channel();
        }
        return $this->channel;
    }

    /**
     * 创建交换机
     * @param  交换机名称
     * @param string $exchange_type 交换机类型 可选 direct fanout  topic headers x-delayed-message（延时队列）
     * @param bool $passive 是否被动申明,true的话会检测是否存在相同的名称的交换机
     * @param bool $durable 是否持久化
     * @param bool $auto_delete 是否自动删除
     * @return array
     */
    public function createExchange($exchange, $exchange_type = 'fanout', $passive = false, $durable = false, $auto_delete = false, $internal = false, $nowait = false, $arguments = array())
    {
        try{
            $this->getChannel()->exchange_declare($exchange, $exchange_type, $passive, $durable, $auto_delete, $internal, $nowait, $arguments);
            echo '创建成功';
        }catch (\Exception $e){
            echo '创建失败'.$e->getMessage();
        }

    }

    /**
     * 创建队列
     * @param
     * @param array $attributes 队列参数
     * @param bool $passive 是否被动申明
     * @param bool $durable 是否持久化
     * @param bool $exclusive 是否排外
     * @param bool $autoDelete 是否自动删除
     * @param bool $nowait
     * @return array 返回创建结果
     */
    public function createQueue($name, $attributes = [], $passive = false, $durable = true, $exclusive = false, $autoDelete = false, $nowait = false)
    {
        try {
            $this->getChannel()->queue_declare($name, $passive, $durable, $exclusive, $autoDelete, $nowait, $attributes);
            echo '创建成功';
        } catch (\Exception $e) {
            echo '创建失败'.$e->getMessage();
        }

    }

    /**
     * 绑定队列
     * @param $queue
     * @param $exchange
     * @param string $routing_key 路由key
     * @param bool $nowait 是否等待返回
     * @param array $arguments 绑定参数
     * @return array 返回操作结果
     */
    public function queueBind($queue, $exchange, $routing_key = '', $nowait = false, $arguments = array())
    {
        try {
            $this->getChannel()->queue_bind($queue, $exchange, $routing_key, $nowait, $arguments);
            echo '绑定成功';
        } catch (\Exception $e) {
            echo '绑定失败'.$e->getMessage();
        }
    }

    /**
     * 发送消息
     * @param $queue
     * @param $body
     * @param array $properties 发送信息属性
     * @param null $exchange 交换机名称
     * @param null $routing_key 路由key
     * @return array 返回结果
     */
    public function sendMessage($queue, $body, $properties = [], $exchange = '', $routing_key = '')
    {
        try {
            $message = new AMQPMessage($body, $properties);
            if (empty($routing_key) && !empty($queue)) $routing_key = $queue;
            $this->getChannel()->basic_publish($message, $exchange, $routing_key);
            echo '消息发送成功';
        } catch (\Exception $e) {
            echo '消息发送失败'.$e->getMessage();
        }
    }

    /**
     * 获取队列信息
     * @param $queue
     * @param int $waitSeconds 等待时间,此参数忽略
     * @return array 返回结果
     */
    public function receiveMessage($queue)
    {
        $receipt = [];
        try {
            $message = $this->getChannel()->basic_get($queue);
            if (!empty($message)) {
                $receipt = [
                    'body' => $message->body,
                    'body_size' => $message->body_size,
                    'is_truncated' => $message->is_truncated,
                    'content_encoding' => $message->content_encoding,
                    'delivery_info' => $message->delivery_info,
                ];
            }
            return $receipt;
        } catch (\Exception $e) {
            echo '获取消息失败'.$e->getMessage();
        }
    }

    /**
     * 消费或者删除消息
     * @param $queue 队列名称
     * @param $receiptHandle delivery_tag值
     * @return array 返回操作结果
     */
    public function deleteMessage($receiptHandle)
    {
        try {
            $this->getChannel()->basic_ack($receiptHandle);
             echo '消费消息成功';
        } catch (\Exception $e) {
            echo '消费消息失败'.$e->getMessage();
        }
    }


}